package com.py.custom.ability.log.kafka;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;
import sun.misc.Signal;

import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class CustomKafkaProducer {

    /**
     * todo 改造成自定义的线程池
     */
    private ExecutorService executorService = Executors.newSingleThreadExecutor();

    private KafkaProducer<String, String> kafkaProducer;

    private String topicName;

    public CustomKafkaProducer(String topicName, String bootstrapServers) {
        this.topicName = topicName;
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.RETRIES_CONFIG, 2);
        kafkaProducer = new KafkaProducer<>(props);
        // 优雅关闭
        shutDown();
    }

    /**
     * 异步发送消息
     *
     * @param logStr
     */
    public void sendLog(String logStr) {
        ProducerRecord<String, String> record = new ProducerRecord<>(topicName, logStr);
        executorService.execute(() -> kafkaProducer.send(record, (recordMetadata, e) -> {
                }
        ));
    }

    public void shutDown() {
        // kill -15 pid
        Signal sg = new Signal("TERM");
        Signal.handle(sg, signal -> {
            // 监听信号量，通过System.exit(0)正常关闭JVM，触发关闭钩子执行收尾工作
            System.exit(0);
        });

        // 执行进程退出前的工作
        // 注意事项：
        // 1.在这里执行的动作不能耗时太久
        // 2.不能在这里再执行注册，移除关闭钩子的操作
        // 3 不能在这里调用System.exit()
        Runtime.getRuntime().addShutdownHook(new Thread(this::close));
    }

    public void close() {
        kafkaProducer.close();
    }
}
